package yuekao6.tm3;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import yuekao6.entity.OrderInfo;
import yuekao6.util.KafkaUtil;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.HashSet;

public class PlaceOrder {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("tms_ods"));
//        streamSource.filter(x -> JSON.parseObject(x).getString("table").equals("order_info")).print();
        DataStreamSource<String> streamSource = env.socketTextStream("hadoop-single", 9999);
        SingleOutputStreamOperator<OrderInfo> orderinfoDS = streamSource.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"))
                .map(new MapFunction<String, OrderInfo>() {
                    @Override
                    public OrderInfo map(String s) throws Exception {
                        String data = JSON.parseObject(s).getString("data");
                        return JSON.parseObject(data, OrderInfo.class);
                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> {
                            try {
                                return new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").parse(event.getCreate_time()).getTime();
                            } catch (ParseException e) {
                                throw new RuntimeException(e);
                            }
                        }));
//        orderinfoDS.print();
        //2.当日下单金额统 计 当日下单总金额，当日下单次数 统计 当日下单次数(10分)
        SingleOutputStreamOperator<String> process = orderinfoDS.windowAll(SlidingEventTimeWindows.of(Time.days(1), Time.hours(1)))
                .process(new ProcessAllWindowFunction<OrderInfo, String, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<OrderInfo, String, TimeWindow>.Context context, Iterable<OrderInfo> iterable, Collector<String> collector) throws Exception {
                        Double price = 0.0;
                        Double sumprice = 0.0;
                        Integer count = 0;
                        for (OrderInfo one : iterable) {
                            if ("60010".equals(one.getStatus())) {
                                price = one.getAmount();
                                sumprice += one.getAmount();
                                count++;
                            }
                        }
                        collector.collect(context.window().getStart()+"------"+context.window().getEnd()+"--------------当日下单金额统:" + price + "当日下单总金额:" + sumprice +  "当日下单总次数:" + count);
                    }
                });
        process.print();

        env.execute();
    }
}
